// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu

#include "coin2/exchange/huobi_futures/symbology/product.h"
#include "coin2/exchange/huobi_futures/feed_v1/processor.h"

#include <set>
#include <string_view>

#include "coin2/base/conversion.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/huobi_futures/feed_v1/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::huobi_futures::feed_v1::impl {
void HuobiFuturesFeedProcessor::OnTopicRecord(const TopicRecordPacket& packet) {
  using coin2::exchange::huobi_futures::symbology::HuobiFuturesProduct;

  upd_mgr_.onBeginPacket(packet);

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(packet.payload);
  }

  simdjson::dom::object obj;
  simdjson::error_code err;
  parser_.parse(packet.payload).get<simdjson::dom::object>().tie(obj, err);

  if (err) {
    Error("[Huobi Futures Feed Processor] err", packet.payload);
    return;
  }

  huobi::feed_v1::HuobiChannel channel;
  std::string_view ch;
  obj["ch"].get<std::string_view>().tie(ch, err);
  if (err) {
    obj["topic"].get<std::string_view>().tie(ch, err);
    if (err) {
      if ((obj["data"].at(0)["volume"].error() != simdjson::NO_SUCH_FIELD) &&
          (obj["data"].at(0)["amount"].error() != simdjson::NO_SUCH_FIELD)) {
        // open interest
        channel.type = "open_interest";
      } else if ((obj["data"].at(0)["index_price"].error() != simdjson::NO_SUCH_FIELD) &&
                 (obj["data"].at(0)["index_ts"].error() != simdjson::NO_SUCH_FIELD)) {
        // open interest
        channel.type = "index";
      } else {
        LOG_FIRST_N(INFO, 1000) << "[Huobi Futures Feed Processor] meta msg for " << packet.topic_name << ": "
                << packet.payload.data();
        return;
      }
    }
  }
  if (!ch.empty()) {
    channel = huobi::feed_v1::HuobiChannel::Parse(std::string(ch));
    simdjson::dom::array data;
    auto err1 = obj["data"].get(data);
    // kline postfix is interval
    if (channel.type.rfind("kline.") == 0 && !err1) {
      // kline rest channel, response channel is kline.1day
      // change to period=1day
      channel.type = channel.type.replace(0, 6, "period=");
    }
  }

  if (!CheckChannelFilter(channel.type)) {
    return;
  }

  std::string_view native_symbol = channel.native_symbol;
  if (channel.type == "trade.detail") {
    if (!upd_mgr_.is_symbol_registered(native_symbol)) {
      return;
    }
    // trades:
    //  symbol, ts
    //   p, q, side
    //   p, q, side
    //   p, q, side
    // ...

    if (parser_config_.has_trade_sampling_frequency_bps()) {
      auto* slim_trade = upd_mgr_.GetSlimTradeMutable(native_symbol);
      int64_t timestamp = obj["ts"].get<int64_t>() * 1'000'000LL;  // ms -> ns
      for (const auto& json_trade : obj["tick"]["data"].get<simdjson::dom::array>()) {
        Trade _trade{};
        _trade.timestamp = timestamp;
        HuobiFuturesFeedUpdateParser::ParseTrade(json_trade, 2, &_trade);
        bool aggregated = slim_trade->TryAggregateSlimTrade(
                packet.timestamp,
                _trade.price,
                _trade.fill_qty,
                _trade.side,
                _trade.timestamp,
                parser_config_.trade_sampling_frequency_bps());
        int64_t timespan = packet.timestamp - slim_trade->LastPublishedTimestamp();
        if (!aggregated ||
            timespan > parser_config_.trade_sampling_time_length()) {
          auto* trade = upd_mgr_.GetTradeMutable();
          CHECK(slim_trade->PopulateTradeAndClear(trade));
          if (trade->fill_qty > 0) {
            upd_mgr_.PublishTrade(native_symbol, false);
          }
        }
      } 
    } else {
      Trade* agg_trade = upd_mgr_.GetTradeMutable();
      agg_trade->Clear();
      Trade trade{};

      agg_trade->timestamp = obj["ts"].get<int64_t>() * 1'000'000LL;  // ms -> ns
      // Huobi trade timestamp is reused, not updated by every trades array.
      trade.timestamp = agg_trade->timestamp;
      for (const auto& trade_json : obj["tick"]["data"].get<simdjson::dom::array>()) {
        HuobiFuturesFeedUpdateParser::ParseTrade(trade_json, 2, &trade);
        if (TryAggregateTrades(agg_trade, trade)) {
          continue;
        }
        if (agg_trade->fill_qty > 0) {
          upd_mgr_.PublishTrade(native_symbol, true);
        }
        *agg_trade = trade;
      }
      if (agg_trade->fill_qty > 0) {
        upd_mgr_.PublishTrade(native_symbol, false);
      }
    }
  } else if (channel.type == "depth.step6" || channel.type == "depth.size_20.high_freq") {
    auto* book = upd_mgr_.GetBookMutable(native_symbol);
    if (book != nullptr) {
      // only parse update book after bbo warm up
      auto& stat = topic_stat_[std::string(native_symbol)];
      if (channel.type == "depth.size_20.high_freq") {
        if (!stat.bbo_received) {
          return;
        }
      }
      if (publish_bbo_book_only_) {
        std::optional<PriceQty> bid_pre = book->Bid0();
        std::optional<PriceQty> ask_pre = book->Ask0();
        HuobiFuturesFeedUpdateParser::ParseBook(book, obj, true);
        std::optional<PriceQty> bid_post = book->Bid0();
        std::optional<PriceQty> ask_post = book->Ask0();
        bool is_bbo_changed = (bid_pre != bid_post) || (ask_pre != ask_post);
        if (is_bbo_changed) {
          upd_mgr_.PublishBook();
        }
      } else {
        HuobiFuturesFeedUpdateParser::ParseBook(book, obj, false);
        upd_mgr_.PublishBook();
      }
      stat.update_received = true;
    }
  } else if (channel.type == "liquidation_orders") {
    simdjson::dom::array orders;
    auto err = obj["data"].get(orders);
    if (err) return;
    for (const auto& order : orders) {
      std::string_view native_symbol = order["contract_code"].get<std::string_view>();
      std::string_view feed_native_symbol = HuobiFuturesProduct::FromNativeStr(
          std::string(native_symbol), packet.timestamp).feed_native_symbol();
      if (!upd_mgr_.is_symbol_registered(feed_native_symbol)) {
        continue;
      }
      auto* liquidation_order = upd_mgr_.GetLiquidationOrderMutable();
      HuobiFuturesFeedUpdateParser::ParseLiquidationOrder(order, liquidation_order);
      upd_mgr_.PublishLiquidationOrder(feed_native_symbol, false);
    }
  } else if (channel.type == "funding_rate") {
    simdjson::dom::array items;
    auto err = obj["data"].get(items);
    if (err) return;
    for (const auto& item : items) {
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        continue;
      }
      auto* funding_rate = upd_mgr_.GetFundingRateMutable();
      HuobiFuturesFeedUpdateParser::ParseFundingRate(item, funding_rate, packet.timestamp);
      upd_mgr_.PublishFundingRate(native_symbol);
    }
  } else if (channel.type == "open_interest") {
    simdjson::dom::array open_interest_array;
    auto err = obj["data"].get(open_interest_array);
    if (err) return;

    int64_t timestamp = obj["ts"].get<int64_t>() * 1'000'000LL;
    for (const auto& msg : open_interest_array) {
      std::string_view native_symbol = msg["contract_code"].get<std::string_view>();
      auto product = HuobiFuturesProduct::FromNativeStrNoThrow(
          std::string(native_symbol), packet.timestamp);
      if (!product) {
        continue;  // unkown symbol, maybe new added symbol
      }

      const std::string& feed_native_symbol = product->feed_native_symbol();
      if (!upd_mgr_.is_symbol_registered(feed_native_symbol)) {
        continue;
      }

      auto* open_interest = upd_mgr_.GetOpenInterestMutable();
      open_interest->timestamp = timestamp;
      open_interest->open_interest_qty = msg["volume"].get<double>();
      upd_mgr_.PublishOpenInterestWithExchangeTimeCheck(feed_native_symbol);
    }
  } else if (channel.type == "index") {
    simdjson::dom::array index_array;
    auto err = obj["data"].get(index_array);
    if (err) return;

    for (const auto& msg : index_array) {
      std::vector<std::string> feed_native_symbols;
      if (msg["symbol"].error() != simdjson::NO_SUCH_FIELD) {
        // futures
        std::string_view native_symbol = msg["symbol"].get<std::string_view>();
        feed_native_symbols = RegisteredFuturesOfCurrency(native_symbol);
      } else if (msg["contract_code"].error() != simdjson::NO_SUCH_FIELD) {
        // USD USDT swap
        std::string_view native_symbol = msg["contract_code"].get<std::string_view>();
        feed_native_symbols.push_back(std::string(native_symbol));
      } else {
        Error("Unkown Index message: ", packet.payload);
        return;
      }
      if (feed_native_symbols.empty()) {
        continue;
      }

      auto* index = upd_mgr_.GetIndexMutable();
      index->timestamp = msg["index_ts"].get<int64_t>() * 1'000'000LL;
      index->price = msg["index_price"].get<double>();
      // some index are queried by https, data may be duplicated
      for (const auto& feed_native_symbol : feed_native_symbols) {
        if (!upd_mgr_.is_symbol_registered(feed_native_symbol)) {
          continue;
        }
        upd_mgr_.PublishIndex(feed_native_symbol);
      }
    }
  } else if (channel.type == "bbo") {
    auto* book = upd_mgr_.GetBookMutable(native_symbol);
    if (book != nullptr) {
      auto& stat = topic_stat_[std::string(native_symbol)];
      if (!stat.bbo_received || !stat.update_received) {
        HuobiFuturesFeedUpdateParser::ParseBbo(book, obj);
        if (parser_config_.has_sampling_frequency_bps()) {
          bool sampled = book->SampleByFrequency(
                  book->Bid0(),
                  book->Ask0(),
                  parser_config_.sampling_frequency_bps());
          if (parser_config_.has_trade_sampling_frequency_bps()) {
            auto* slim_trade = upd_mgr_.GetSlimTradeMutable(native_symbol);
            int64_t timespan = packet.timestamp - slim_trade->LastPublishedTimestamp();
            if (sampled ||
                timespan > parser_config_.trade_sampling_time_length()) {
              auto* trade = upd_mgr_.GetTradeMutable();
              if (slim_trade->PopulateTradeAndClear(trade)) {
                upd_mgr_.PublishTrade(native_symbol, false);
              }
            }
          }
          if (sampled) {
            upd_mgr_.PublishBook();
          }
        } else {
          upd_mgr_.PublishBook();
        }
        stat.bbo_received = true;
      }
    }
  } else if (channel.type.rfind("kline") == 0) {
    // kline websocket
    if (!upd_mgr_.is_symbol_registered(native_symbol)) {
      return;
    }

    // market.BTC-USD.kline.1day
    std::string_view period = ch.substr(ch.find_last_of('.') + 1);
    auto interval = HuobiFeedUpdateParser::ParseKlineInterval(period);
    auto* klines = upd_mgr_.GetKlineMutable(native_symbol, interval);
    if (klines == nullptr) return;

    int64_t ts = obj["ts"].get<int64_t>() * 1'000'000LL;

    simdjson::dom::object tick;
    auto err_tick = obj["tick"].get(tick);
    bool to_publish = false;
    if (!err_tick){
      // from websocket
      klines->UpdateClear();
      to_publish = HuobiFuturesFeedUpdateParser::ParseKline(klines, tick, interval, ts);
    } else {
      NOTREACHED() << "Kline invalid" << obj;
      return;
    }

    if (to_publish) upd_mgr_.PublishKline(native_symbol);
  } else if (channel.type.rfind("period=") == 0) {
    // kline rest
    if (!upd_mgr_.is_symbol_registered(native_symbol)) {
      return;
    }

    // market.BTC-USD.kline.1day
    std::string_view period = ch.substr(ch.find_last_of('.') + 1);
    auto interval = HuobiFeedUpdateParser::ParseKlineInterval(period);
    auto* klines = upd_mgr_.GetKlineMutable(native_symbol, interval);
    if (klines == nullptr) return;

    int64_t ts = obj["ts"].get<int64_t>() * 1'000'000LL;

    simdjson::dom::array data;
    auto err_data = obj["data"].get(data);
    bool to_publish = false;
    if (!err_data) {
      // from rest api
      std::optional<int64_t> first_open_timestamp;
      bool asc = true;
      if (data.size() >= 2) {
        first_open_timestamp = data.at(0)["id"].get<int64_t>() * 1'000'000'000LL;
        int64_t second_open_timestamp = data.at(1)["id"].get<int64_t>() * 1'000'000'000LL;
        asc = (first_open_timestamp < second_open_timestamp);
      }

      if (klines->Size() == 0) {
        klines->SnapshotClear();
      } else if (first_open_timestamp &&
                 first_open_timestamp < klines->RBegin()->open_timestamp) {
        klines->SnapshotClear();
      } else {
        klines->UpdateClear();
      }

      if (asc) {
        for (const auto& kline_obj : data) {
          bool freezed = HuobiFuturesFeedUpdateParser::ParseKline(klines, kline_obj,
              interval, ts);
          to_publish = to_publish || freezed;
        }
      } else {
        for (int i = data.size() - 1; i >= 0; i--) {
          const auto& kline_obj = data.at(i);
          bool freezed = HuobiFuturesFeedUpdateParser::ParseKline(klines, kline_obj,
              interval, ts);
          to_publish = to_publish || freezed;
        }
      }
    } else {
      NOTREACHED() << "Kline invalid" << obj;
      return;
    }

    if (to_publish) upd_mgr_.PublishKline(native_symbol);
  } else {
    Error("[Huobi Futures Feed Processor] unknown channel: " + channel.type, packet.payload);
  }

  upd_mgr_.onEndPacket(packet);
}

std::vector<std::string> HuobiFuturesFeedProcessor::RegisteredFuturesOfCurrency(
    std::string_view native_symbol) {
  static std::set<base::symbology::ContractType> futures = {
      ContractType::WEEK,
      ContractType::NEXT_WEEK,
      ContractType::MONTH,
      ContractType::NEXT_MONTH,
      ContractType::QUARTER,
      ContractType::NEXT_QUARTER};

  std::vector<std::string> symbols;
  for (const auto& product : upd_mgr_.products()) {
    if (product->market() == coin2::exchange::base::market::MarketType::Futures) {
      const symbology::HuobiFuturesProduct& fproduct =
          dynamic_cast<const symbology::HuobiFuturesProduct&>(*product);
      if (futures.find(fproduct.contract_type()) != futures.end() &&
          fproduct.base().native_currency() == native_symbol) {
        symbols.push_back(fproduct.feed_native_symbol());
      }
    }
  }

  return symbols;
}

}  // namespace coin2::exchange::huobi_futures::feed_v1::impl
